package com.avo.ramcache.persist.queue;

import com.avo.ramcache.config.IEntity;
import com.avo.ramcache.orm.Accessor;
import com.avo.ramcache.persist.Element;
import com.avo.ramcache.persist.EventType;
import com.avo.ramcache.persist.Listener;
import com.avo.ramcache.persist.Persister;
import com.avo.ramcache.model.ConcurrentHashSet;

import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.LinkedBlockingQueue;

public class QueuePersister implements Persister {

    private boolean initialize;
    private QueueConsumer consumer;
    private boolean inhibitionRepeat;
    private BlockingQueue<Element> queue;
    private final ConcurrentHashSet<String> updating = new ConcurrentHashSet<>();
    private final ConcurrentHashMap<Class<? extends IEntity<?>>, Listener> listeners = new ConcurrentHashMap<>();
    private volatile boolean stop;

    @Override
    public void initialize(String name, Accessor accessor, String config) {
        if (this.initialize) {
            throw new RuntimeException("重复初始化异常");
        }
        String[] arr = config.split(":");
        int queueSize = Integer.parseInt(arr[0]);
        if (queueSize >= 0) {
            queue = new ArrayBlockingQueue<>(queueSize);
        } else {
            queue = new LinkedBlockingQueue<>();
        }
        inhibitionRepeat = Boolean.parseBoolean(arr[1]);
        consumer = new QueueConsumer(name, accessor, queue, this);
        initialize = true;
    }

    @Override
    public void put(Element element) {
        if (this.stop) {
            throw new RuntimeException("实体队列停止更新，不接受新元素");
        }
        String identity = element.getIdentity();
        if (inhibitionRepeat && element.getType() == EventType.UPDATE) {
            if (updating.contains(identity)) {
                return;
            }
            updating.add(identity);
        }
        queue.add(element);
    }

    @Override
    public void addListener(Class<? extends IEntity<?>> clz, Listener listener) {
        if (listener == null) {
            throw new RuntimeException("监听器为空");
        }
        this.listeners.put(clz, listener);
    }

    @Override
    public Listener getListener(Class<? extends IEntity<?>> clz) {
        return this.listeners.get(clz);
    }

    @Override
    public void shutdown() {
        this.stop = true;
        consumer.stop();
    }

    @Override
    public Map<String, String> getInfo() {
        HashMap<String, String> result = new HashMap<>();
        result.put("size", Integer.toString(queue.size()));
//        result.put("state", consumer.getState().name());
        return result;
    }


}
